package com.wykj.framework.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;


public class KafKaProducerUtil {
	private static String KAFKA_METADATA_BROKER_LIST="192.168.1.222:9092";
	private static String KAFKA_KEY_SERIALIZER_CLASS="kafka.serializer.StringEncoder";
	private static String KAFKA_SERIALIZER_CLASS="kafka.serializer.StringEncoder";
	private static String KAFKA_REQUEST_REQUIRED_ACKS="1";
	private static String KAFKA_COMPRESSION_CODEC="snappy";
	private static String KAFKA_MESSAGE_SEND_MAX_RETRIES="2";
	private static String KAFKA_RECONNECT_BACKOFF_MS="100";
	private static String KAFKA_REQUEST_TIMEOUT_MS="10000";
	private static String KAFKA_PRODUCER_TYPE="sync"; //同步处理
	private static String KAFKA_QUEUE_BUFFERING_MAX_MS=""; //消息缓存最长时间 异步方式下配置
	private static String KAFKA_QUEUE_BUFFERING_MAX_MESSAGES="";//缓存消息的最大值
	private static String KAFKA_QUEUE_ENQUEUE_TIMEOUT_MS="";//消息进入队列的的等待时间
	private static String KAFKA_BATCH_NUM_MESSAGES="";//批量消息条数
	private static String KAFKA_SEND_BUFFER_BYTES="";//socket缓存大小

	private static Producer<String, String> producer;

	private static Producer<String, String> getProducer(){
		if(producer==null){
			synchronized (KafKaProducerUtil.class) {
				Properties pro=new Properties();
				pro.setProperty("metadata.broker.list",KAFKA_METADATA_BROKER_LIST);
				pro.setProperty("request.required.acks", KAFKA_REQUEST_REQUIRED_ACKS);
				pro.setProperty("producer.type", KAFKA_PRODUCER_TYPE);
				pro.setProperty("serializer.class", KAFKA_SERIALIZER_CLASS);
				pro.setProperty("request.timeout.ms", KAFKA_REQUEST_TIMEOUT_MS);
				pro.setProperty("key.serializer.class", KAFKA_KEY_SERIALIZER_CLASS);
				pro.setProperty("compression.codec", KAFKA_COMPRESSION_CODEC);
				pro.setProperty("message.send.max.retries", KAFKA_MESSAGE_SEND_MAX_RETRIES);
				pro.setProperty("retry.backoff.ms", KAFKA_RECONNECT_BACKOFF_MS);
				ProducerConfig config=new ProducerConfig(pro);
				producer=new  Producer<String, String>(config);
			}
		}
		return producer;
	}


	public static void sendMassage(String topic,String key,String message){
		Producer<String, String> producer=getProducer();
		KeyedMessage<String, String> km=new KeyedMessage<String, String>(topic, key, message);
		producer.send(km);
	}


	public static void sendListMassage(String topic,Map<String,String> msg){
		Producer<String, String> producer=getProducer();
		List<KeyedMessage<String, String>> list=new ArrayList<KeyedMessage<String, String>>();
		for(String k : msg.keySet()){
			KeyedMessage<String, String> km=new KeyedMessage<String, String>(topic, k, msg.get(k));
			list.add(km);
		}
		producer.send(list);
	}
}
